package com.atguigu.wordCount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author gmd
 * @desc DataSet API实现WordCount（不推荐）
 * @since 2024-11-21 11:52:19
 */
public class BatchOfReadFile {

    public static void main(String[] args) throws Exception {
        // 创建执行环境，批处理执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 读取数据：从文件中读取（文件内容就是按空格分隔开的单词）
        DataSource<String> lineDS = env.readTextFile("input/word.txt");

        // 按空格切分单词、转换成Tuple2(word, 1)对象
        FlatMapOperator<String, Tuple2<String, Integer>> wordAndOne = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                // 按照空格切分单词
                String[] words = value.split(" ");
                // 遍历
                for (String word : words) {
                    // 将单词转换为Tuple2(word, 1)
                    Tuple2<String, Integer> wordTuple2 = Tuple2.of(word, 1);
                    //使用 Collector 向下游发送数据
                    out.collect(wordTuple2);
                }
            }
        });

        // 分组
        // 0是位置，表示按照word分组
        UnsortedGrouping<Tuple2<String, Integer>> wordGroupBy = wordAndOne.groupBy(0);

        // 各分组内聚合
        // 1是位置，表示第二个元素进行sum求和
        AggregateOperator<Tuple2<String, Integer>> wordSum = wordGroupBy.sum(1);
        // 输出
        wordSum.print();
    }

    // Lambda表达式的精简写法。重点就是需要显式指定被Lambda隐去的元素类型。

    /**
     * Flink还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应的序列化器和反序列化器。
     * 但是，由于Java中泛型擦除的存在，在某些特殊情况下（比如Lambda表达式中），自动提取的信息是不够精细，
     * 只告诉Flink当前的元素由“船头、船身、船尾”构成，根本无法重建出“大船”的模样；
     * 这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。
     */
    public static void main2(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> lineDS = env.readTextFile("./input/word.txt");

        // 必须显式指定返回类型
        AggregateOperator<Tuple2<String, Integer>> result = lineDS
                // 必须显式指定返回对象out的类型，这里显式指定为Collector<Tuple2<String, Integer>>类型
                .flatMap((String ele, Collector<Tuple2<String, Integer>> out) -> {
                    String[] words = ele.split(" ");
                    for (String word : words) {
                        out.collect(Tuple2.of(word, 1));
                    }
                })
                // 必须显式指定flatMap后的子元素类型，这里显式指定为Tuple2<String, Integer>类型
               // .returns(Types.TUPLE(Types.STRING, Types.INT))
                .returns(new TypeHint<Tuple2<String, Integer>>() {})
                .groupBy(0)
                .sum(1);

        // 打印结果，使用System.out.println()只能打印对象的内存地址
        result.print();
    }

}
